package com.nbcio.iot.protocol.mqtt;

import com.nbcio.iot.config.IotMqttProperties;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiConsumer;

/**
 * MQTT 客户端管理器（最小实现）
 */
@Slf4j
@Component
@DependsOn("embeddedMqttBroker")
@RequiredArgsConstructor
public class MqttClientManager {

	private final IotMqttProperties properties;

	private MqttClient client;
	private final AtomicBoolean connected = new AtomicBoolean(false);
	private BiConsumer<String, String> messageListener;

	public void setMessageListener(BiConsumer<String, String> listener) {
		this.messageListener = listener;
	}

	@PostConstruct
	public void init() {
		try {
			client = new MqttClient(properties.getBrokerUrl(), properties.getClientId(), new MemoryPersistence());
			client.setCallback(new MqttCallback() {
				@Override
				public void connectionLost(Throwable cause) {
					connected.set(false);
					log.warn("MQTT connection lost: {}", cause == null ? null : cause.getMessage());
				}

				@Override
				public void messageArrived(String topic, MqttMessage message) {
					String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
					// 关键信息日志，便于排查是否收到了消息
					int len = payload.length();
					String preview = len > 256 ? payload.substring(0, 256) + "..." : payload;
					log.info("[MQTT-CLIENT] RECEIVED topic={}, qos={}, bytes={}, payloadPreview={}", topic, message.getQos(), message.getPayload() == null ? 0 : message.getPayload().length, preview);
					if (messageListener != null) {
						messageListener.accept(topic, payload);
					}
				}

				@Override
				public void deliveryComplete(org.eclipse.paho.client.mqttv3.IMqttDeliveryToken token) { }
			});

			MqttConnectOptions options = new MqttConnectOptions();
			options.setCleanSession(properties.isCleanSession());
			options.setKeepAliveInterval(properties.getKeepAlive());
			options.setAutomaticReconnect(properties.isReconnect());
			if (properties.getUsername() != null) {
				options.setUserName(properties.getUsername());
			}
			if (properties.getPassword() != null) {
				options.setPassword(properties.getPassword().toCharArray());
			}

			// Wait for broker reachability (up to ~5 seconds)
			waitForBrokerReachable(properties.getBrokerUrl(), 5_000);

			log.info("MQTT connecting to {} as {} (autoReconnect={})", properties.getBrokerUrl(), properties.getClientId(), properties.isReconnect());
			client.connect(options);
			connected.set(true);
			log.info("[MQTT-CLIENT] CLIENT HEALTH: OK (connected to {})", properties.getBrokerUrl());

			// 默认订阅通配主题
			if (properties.getDefaultTopicPattern() != null) {
				subscribe(properties.getDefaultTopicPattern());
			}
		} catch (Exception e) {
			log.error("MQTT init error (brokerUrl={})", properties.getBrokerUrl(), e);
		}
	}

	public void subscribe(String topic) {
		try {
			if (client != null && connected.get()) {
				client.subscribe(topic, 1);
				log.info("[MQTT-CLIENT] SUBSCRIBED topic={}", topic);
			} else if (client != null) {
				client.subscribe(topic, 1);
				log.info("[MQTT-CLIENT] SUBSCRIBED (queued/auto-reconnect) topic={}", topic);
			}
		} catch (Exception e) {
			log.error("MQTT subscribe error", e);
		}
	}

	public void publish(String topic, String payload) {
		try {
			if (client != null && connected.get()) {
				MqttMessage message = new MqttMessage(payload.getBytes(StandardCharsets.UTF_8));
				message.setQos(1);
				client.publish(topic, message);
				log.info("MQTT published: {} => {}", topic, payload);
			} else if (client != null) {
				MqttMessage message = new MqttMessage(payload.getBytes(StandardCharsets.UTF_8));
				message.setQos(1);
				client.publish(topic, message);
				log.info("MQTT published (auto-reconnect path): {} => {}", topic, payload);
			}
		} catch (Exception e) {
			log.error("MQTT publish error", e);
		}
	}

	@PreDestroy
	public void destroy() {
		try {
			if (client != null && client.isConnected()) {
				client.disconnect();
			}
		} catch (Exception ignored) { }
	}

	private void waitForBrokerReachable(String brokerUrl, int maxWaitMs) {
		try {
			URI uri = URI.create(brokerUrl);
			String host = uri.getHost() == null ? "127.0.0.1" : uri.getHost();
			int port = uri.getPort() == -1 ? 1883 : uri.getPort();
			long deadline = System.currentTimeMillis() + maxWaitMs;
			while (System.currentTimeMillis() < deadline) {
				try (Socket s = new Socket()) {
					s.connect(new InetSocketAddress(host, port), 500);
					return;
				} catch (Exception ignore) {
					try { Thread.sleep(200); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); return; }
				}
			}
			log.warn("MQTT broker not reachable within {} ms ({}:{}) — proceeding, autoReconnect may recover", maxWaitMs, host, port);
		} catch (Exception ex) {
			log.warn("Failed to parse broker url {}, proceeding without reachability check", brokerUrl);
		}
	}
} 